﻿using Confluent.Kafka;
using Confluent.Kafka.Admin;
using System;
using System.Threading.Tasks;

namespace AdminClient
{
	class Program
	{
		static string ToString(int[] array) => $"[{string.Join(", ", array)}]";

		static void ListGroups(string bootstrapServers)
		{
			using (var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrapServers }).Build())
			{
				// Warning: The API for this functionality is subject to change.
				var groups = adminClient.ListGroups(TimeSpan.FromSeconds(10));
				Console.WriteLine($"Consumer Groups:");
				foreach (var g in groups)
				{
					Console.WriteLine($"  Group: {g.Group} {g.Error} {g.State}");
					Console.WriteLine($"  Broker: {g.Broker.BrokerId} {g.Broker.Host}:{g.Broker.Port}");
					Console.WriteLine($"  Protocol: {g.ProtocolType} {g.Protocol}");
					Console.WriteLine($"  Members:");
					foreach (var m in g.Members)
					{
						Console.WriteLine($"    {m.MemberId} {m.ClientId} {m.ClientHost}");
						Console.WriteLine($"    Metadata: {m.MemberMetadata.Length} bytes");
						Console.WriteLine($"    Assignment: {m.MemberAssignment.Length} bytes");
					}
				}
			}
		}

		static void PrintMetadata(string bootstrapServers)
		{
			using (var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrapServers }).Build())
			{
				// Warning: The API for this functionality is subject to change.
				var meta = adminClient.GetMetadata(TimeSpan.FromSeconds(20));
				Console.WriteLine($"{meta.OriginatingBrokerId} {meta.OriginatingBrokerName}");
				meta.Brokers.ForEach(broker =>
					Console.WriteLine($"Broker: {broker.BrokerId} {broker.Host}:{broker.Port}"));

				meta.Topics.ForEach(topic =>
				{
					Console.WriteLine($"Topic: {topic.Topic} {topic.Error}");
					topic.Partitions.ForEach(partition =>
					{
						Console.WriteLine($"  Partition: {partition.PartitionId}");
						Console.WriteLine($"    Replicas: {ToString(partition.Replicas)}");
						Console.WriteLine($"    InSyncReplicas: {ToString(partition.InSyncReplicas)}");
					});
				});
			}
		}

		static async Task CreateTopicAsync(string bootstrapServers, string topicName)
		{
			using (var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrapServers }).Build())
			{
				try
				{
					await adminClient.CreateTopicsAsync(new TopicSpecification[] {
						new TopicSpecification { Name = topicName, ReplicationFactor = 1, NumPartitions = 1 } });
				}
				catch (CreateTopicsException e)
				{
					Console.WriteLine($"An error occured creating topic {e.Results[0].Topic}: {e.Results[0].Error.Reason}");
				}
			}
		}

		public static async Task Main(string[] args)
		{
			string brokerList = "39.96.82.51:9093,47.95.2.2:9092,39.96.34.52:9092";
			string flag = "metadata";
			switch (flag)
			{
				case "library-version":
					Console.WriteLine($"librdkafka Version: {Library.VersionString} ({Library.Version:X})");
					Console.WriteLine($"Debug Contexts: {string.Join(", ", Library.DebugContexts)}");
					break;
				case "list-groups":
					ListGroups(brokerList);
					break;
				case "metadata":
					PrintMetadata(brokerList);
					break;
				case "create-topic":
					await CreateTopicAsync(brokerList, "topic");
					break;
			}
			Console.ReadKey();
		}

	}
}
